//! 零锁网络层实现
//! 
//! 真正的单线程Redis架构：
//! 1. 单个Redis工作线程处理所有命令
//! 2. 网络连接通过无锁队列发送命令
//! 3. 零锁竞争，极致性能

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::io;
use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use async_trait::async_trait;
use crate::core::network::AsyncNetworkStream;
use crate::protocol::RespParser;
use crate::storage::{SingleThreadRedisServer};
use crate::event::ConnectionId;
use tracing::{info, warn, error};

/// 零锁Tokio网络流适配器
#[derive(Debug)]
pub struct ZeroLockTokioStream {
    stream: TcpStream,
    remote_addr: String,
}

impl ZeroLockTokioStream {
    pub fn from_tokio_stream(tokio_stream: TcpStream) -> io::Result<Self> {
        let remote_addr = tokio_stream.peer_addr()
            .map(|addr| addr.to_string())
            .unwrap_or_else(|_| "unknown".to_string());
        
        Ok(Self {
            stream: tokio_stream,
            remote_addr,
        })
    }
}

#[async_trait]
impl AsyncNetworkStream for ZeroLockTokioStream {
    async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.stream.read(buf).await
    }
    
    async fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.stream.write(buf).await
    }
    
    async fn flush(&mut self) -> io::Result<()> {
        self.stream.flush().await
    }
    
    fn is_connected(&self) -> bool {
        self.stream.peer_addr().is_ok()
    }
    
    fn remote_addr(&self) -> Option<String> {
        Some(self.remote_addr.clone())
    }
    
    async fn shutdown(&mut self) -> io::Result<()> {
        self.stream.shutdown().await
    }
    
    fn network_type(&self) -> &'static str {
        "ZeroLock-TCP"
    }
}

/// 零锁连接处理器
/// 
/// 核心特性：
/// 1. 每个连接独立处理协议解析
/// 2. 所有命令发送到共享的单线程Redis服务器
/// 3. 零锁设计，极致性能
pub struct ZeroLockConnectionHandler {
    connection_id: ConnectionId,
    redis_server: Arc<SingleThreadRedisServer>,
    resp_parser: RespParser,
}

impl ZeroLockConnectionHandler {
    pub fn new(connection_id: ConnectionId, redis_server: Arc<SingleThreadRedisServer>) -> Self {
        Self {
            connection_id,
            redis_server,
            resp_parser: RespParser::new(),
        }
    }

    /// 处理连接
    pub async fn handle_connection(&mut self, mut stream: ZeroLockTokioStream) -> io::Result<()> {
        let mut buffer = vec![0u8; 8192];
        
        info!("🔗 连接 {} 开始处理", self.connection_id);
        
        while stream.is_connected() {
            match stream.read(&mut buffer).await {
                Ok(0) => {
                    // 连接正常关闭
                    break;
                }
                Ok(n) => {
                    let data = &buffer[..n];
                    self.resp_parser.feed_data(data);
                    
                    // 解析所有完整的命令
                    while let Ok(Some(resp_value)) = self.resp_parser.parse_value() {
                        match self.resp_parser.resp_to_command(resp_value) {
                            Ok(command) => {
                                let command_name = command.name().to_string();
                                let args = command.to_args();
                                
                                // 发送命令到单线程Redis服务器（零锁）
                                match self.redis_server.execute_command_sync(command_name, args) {
                                    Ok(result) => {
                                        let response = result.to_resp().into_bytes();
                                        if let Err(e) = stream.write(&response).await {
                                            error!("连接 {} 写入响应失败: {}", self.connection_id, e);
                                            return Err(e);
                                        }
                                        if let Err(e) = stream.flush().await {
                                            error!("连接 {} flush失败: {}", self.connection_id, e);
                                            return Err(e);
                                        }
                                    }
                                    Err(e) => {
                                        error!("连接 {} 执行命令失败: {}", self.connection_id, e);
                                        let error_response = format!("-ERR {e}\r\n");
                                        let _ = stream.write(error_response.as_bytes()).await;
                                        let _ = stream.flush().await;
                                    }
                                }
                            }
                            Err(e) => {
                                warn!("连接 {} 协议解析失败: {:?}", self.connection_id, e);
                                let error_response = "-ERR Protocol parsing error\r\n";
                                let _ = stream.write(error_response.as_bytes()).await;
                                let _ = stream.flush().await;
                                break;
                            }
                        }
                    }
                }
                Err(e) => {
                    error!("连接 {} 读取失败: {}", self.connection_id, e);
                    return Err(e);
                }
            }
        }
        
        info!("🔌 连接 {} 已断开", self.connection_id);
        Ok(())
    }
}

/// 零锁Redis服务器
/// 
/// 架构设计：
/// 1. 启动单个Redis工作线程
/// 2. 接受多个网络连接
/// 3. 每个连接发送命令到Redis工作线程
/// 4. 完全无锁设计
pub struct ZeroLockRedisServer {
    bind_addr: String,
    connection_counter: AtomicU64,
    redis_server: Arc<SingleThreadRedisServer>,
}

impl ZeroLockRedisServer {
    pub fn new(bind_addr: String) -> Self {
        Self {
            bind_addr,
            connection_counter: AtomicU64::new(0),
            redis_server: Arc::new(SingleThreadRedisServer::new()),
        }
    }
    
    pub async fn start(&mut self) -> io::Result<()> {
        // 启动单线程Redis服务器
        Arc::get_mut(&mut self.redis_server).unwrap().start();
        
        let listener = TcpListener::bind(&self.bind_addr).await?;
        
        info!("🚀 零锁Redis服务器启动");
        info!("📍 监听地址: {}", self.bind_addr);
        info!("⚡ 架构: 单线程Redis + 零锁网络");
        
        loop {
            match listener.accept().await {
                Ok((stream, _addr)) => {
                    let connection_id = self.connection_counter.fetch_add(1, Ordering::SeqCst);
                    let redis_server = self.redis_server.clone();
                    
                    tokio::spawn(async move {
                        let network_stream = match ZeroLockTokioStream::from_tokio_stream(stream) {
                            Ok(s) => s,
                            Err(e) => {
                                error!("创建网络流失败: {}", e);
                                return;
                            }
                        };
                        
                        let mut handler = ZeroLockConnectionHandler::new(connection_id, redis_server);
                        
                        if let Err(e) = handler.handle_connection(network_stream).await {
                            error!("连接 {} 处理失败: {}", connection_id, e);
                        }
                    });
                }
                Err(e) => {
                    error!("接受连接失败: {}", e);
                    continue;
                }
            }
        }
    }
    
    /// 获取Redis服务器统计信息
    pub fn get_stats(&self) -> &crate::storage::SingleThreadRedisStats {
        self.redis_server.get_stats()
    }
    
    /// 获取当前QPS
    pub fn get_current_qps(&self) -> u64 {
        self.redis_server.get_current_qps()
    }
}

/// 启动零锁Redis服务器
pub async fn start_zero_lock_redis_server(bind_addr: &str) -> io::Result<()> {
    let mut server = ZeroLockRedisServer::new(bind_addr.to_string());
    server.start().await
}